package com.qf.pubsub;

import com.qf.util.ConnUtil;
import com.rabbitmq.client.*;
import org.junit.Test;

import java.io.IOException;

public class Publisher {

    @Test
    public void publish() throws Exception {
        //1. 获取connection对象
        Connection connection = ConnUtil.getConnection();

        //2. 创建Channel对象
        Channel channel = connection.createChannel();

        //3. 创建交换机
        channel.exchangeDeclare("pubsub-exchange", BuiltinExchangeType.FANOUT);

        //4. 创建队列2个
        channel.queueDeclare("pubsub-queue1", true, false, false, null);
        channel.queueDeclare("pubsub-queue2", true, false, false, null);

        //5. 交换机绑定队列
        channel.queueBind("pubsub-queue1", "pubsub-exchange", "432423432");
        channel.queueBind("pubsub-queue2", "pubsub-exchange", "34534654654645");


        // confirm机制开启
        channel.confirmSelect();
        // 添加了回调方法
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("消息已经送达到交换机");
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("消息丢了,没送到交换机");
            }
        });

        channel.addReturnListener(new ReturnListener() {
            @Override
            public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消息没有路由到队列");
            }
        });

        //6. 发送消息到交换机
        AMQP.BasicProperties props = new AMQP.BasicProperties().builder().deliveryMode(2).build();// deliveryMode设置为2，代表队列中的消息需要持久化
        String msg = "publish/subscribe!!";
        channel.basicPublish("pubsub-exchange", "",true, props, msg.getBytes());


        System.in.read();

        // confirm机制确认
        // 需要等待
        /*if (channel.waitForConfirms()) {

        }*/

        // 需要等待
        /*try {
            channel.waitForConfirmsOrDie();
        } catch (IOException e) {
            e.printStackTrace();
        }*/


    }

}
